package com.edata.bigdata.hdfs

import com.edata.bigdata.annotations.Edata_Loader
import com.edata.bigdata.basic.Loader
import com.edata.bigdata.utils.DataFrameUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.ipc.StandbyException
import org.apache.spark.sql.{DataFrame, SparkSession}

import java.io.FileNotFoundException
import scala.util.control.Breaks.breakable

@Edata_Loader(target = "HDFSLOADER")
class HdfsLoader extends Loader with HdfsConnector {
  override var sourceType: String = "HDFS"
  override var session: SparkSession = _
  var client: FileSystem = _
  var client_capacity: Long = -1L
  var active_namenode: String = _


  override def find(args: String*): DataFrame = {
    if(client==null||client.getStatus().getCapacity<0){
      createHdfsClient()
    }
    if(args.isEmpty){
      LOGGER.info(s"args should not be empty")
    }
    val filePath = active_namenode + args(0)
    if(!fileExisted(filePath)){
      throw new FileNotFoundException(s"file does not exist: ${filePath}")
    }
    val rdd =session.sparkContext.textFile(filePath)
    val colNames = args(2).split(args(1))
    val colTypes = DataFrameUtils.parseDataTypeByStr(args(3).split(args(1)))
    val nullables = DataFrameUtils.parseBooleanByStr(args(4).split(args(1)))
    DataFrameUtils.createDataFrame(session,rdd,args(1),colNames,colTypes,nullables)
  }

  def createHdfsClient(): Unit = {
    val conf = new Configuration()
    breakable {
      val nns = HDFS_ENTRYPOINT.split(",")
      for (nn <- nns) {
        conf.set("fs.defaultFS", URI_PREFIX + nn)
        client = FileSystem.get(conf)
        try {
          client_capacity = client.getStatus.getCapacity
          active_namenode = URI_PREFIX + nn
          LOGGER.info(s"${nn} is active")
        } catch {
          case ex: StandbyException => {
            LOGGER.error(s"${nn} is not active")
          }
        }
      }
    }
  }

  def fileExisted(path:String):Boolean={
    val file = new Path(path)
    client.exists(file)
  }


}
